package com.zqxq.lottery.mq;

import com.alibaba.fastjson.JSON;
import com.zqxq.lottery.common.constant.HuaYangConstant;
import com.zqxq.lottery.common.mq.ExpirationMessagePostProcessor;
import org.springframework.amqp.core.MessagePostProcessor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * desc: RabbitMQ生产者
 * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调。
 * ReturnCallback接口用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调。
 */
@Component
@Slf4j
public class Sender {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    /**
     * 发送消息
     * @param queue     队列
     * @param message   消息内容
     */
    public void send(String queue,Object message){
        log.info("发送消息开始");
        log.info("消息队列{}:数据{}", queue,JSON.toJSONString(message));
        rabbitTemplate.convertAndSend(queue,message);
        log.info("发送消息结束");
    }

    /**
     * 发送消息
     * @param queue     队列
     * @param message   消息内容
     * @param ttl   消息过期时间，单位毫秒
     */
    public void send(String queue,Object message, Long ttl){
        rabbitTemplate.convertAndSend(queue,message, new ExpirationMessagePostProcessor(1L));
    }

    public void send(String queueName, String msg, long times) {
        DLXMessage dlxMessage = new DLXMessage(queueName,msg,times);
        MessagePostProcessor processor = new MessagePostProcessor(){
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration(times + "");
                return message;
            }
        };
        dlxMessage.setExchange(HuaYangConstant.DEFAULT_EXCHANGE);
        rabbitTemplate.convertAndSend(HuaYangConstant.DEFAULT_EXCHANGE,HuaYangConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME, JSONUtils.toJson(dlxMessage), processor);
    }

    /**
     * 转换并发送消息,且等待消息者返回响应消息。
     * @param queue
     * @param message
     */
    public void sendAndReceive(String queue,Object message){
        log.info("发送消息开始");
        Object receiveMessage =rabbitTemplate.convertSendAndReceive(queue,message);
        log.info("发送消息结束");
        log.info("消费者响应消息:{} 处理完成", JSON.toJSONString(receiveMessage));
    }

    public void testSendAndReceiveMessageTTL(String queue,Object message){
        rabbitTemplate.convertAndSend(queue,message, new ExpirationMessagePostProcessor(10000L));
    }
}
